package com.example.demo.consumer;


import com.example.demo.bean.Account;
import com.example.demo.bean.Order;
import com.example.demo.exception.AccountNotExistedException;
import com.example.demo.service.AccountService;
import com.example.demo.task.TransferTask;
import com.example.demo.untils.RedisUtil;
import com.github.phantomthief.collection.BufferTrigger;
import com.github.phantomthief.pool.KeyAffinityExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
@RocketMQMessageListener(topic = "hotAccount",consumerGroup = "order-consumer-group")
public class HotAccountConsumer implements RocketMQListener<Order> {


    @Autowired
    private RedisUtil redisUtil;

    BufferTrigger<Order> bufferTrigger = BufferTrigger.<Order>batchBlocking()
            .bufferSize(1000)
            .batchSize(10)
            .linger(Duration.ofSeconds(60))
            .setConsumerEx(this::consume)
            .build();
    @Autowired
    private AccountService accountService;
    private void consume(List<Order> orders) throws AccountNotExistedException {
        //亲缘线程池
        KeyAffinityExecutor executor =
                KeyAffinityExecutor.newSerializingExecutor(8, 200, "MY-POOL");
        double money = 0;
        boolean flag = false;
        for(int i = 0 ;i < orders.size();i++){
            Order order = orders.get(i);
            money += order.getMoney();
            String orderId = order.getOrderId();
            Account fromAccount = accountService.getById(order.getMerchantId());
            if(fromAccount == null) throw new AccountNotExistedException(order.getMerchantId());
            //创建转账任务
            if(i == orders.size() - 1) flag = true;
            TransferTask task = new TransferTask(fromAccount,money,orderId,flag);
            executor.executeEx(order.getMerchantId(),task);
        }
    }

    @Override
    public void onMessage(Order order) {
        boolean exists = redisUtil.set("orderId" + order.getOrderId(),1,1000, TimeUnit.SECONDS);
        if(exists){
            //消费记录已经存在跳过当前消息
            return;
        }
        bufferTrigger.enqueue(order);

    }
}
